package com.filter;

import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * 利用Activate拓展点对消费者进行拦截
 * @author lihongfeng
 */
@Activate(group = {CommonConstants.CONSUMER})
public class TPMonitorFilter implements Filter {
    private final Map<String, DelayQueue<SpentTime>> mappedMethodSpentTime = new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
    {


        scheduledExecutorService.scheduleAtFixedRate(() -> {
            System.out.println(LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME));
            for (Map.Entry<String, DelayQueue<SpentTime>> entry : mappedMethodSpentTime.entrySet()) {
                final DelayQueue<SpentTime> delayQueue = entry.getValue();
                // 移除大于60秒的数据
                delayQueue.removeIf(spentTime -> spentTime.getDelay(TimeUnit.MILLISECONDS) <= 0);
                List<SpentTime> spentTimeList = delayQueue.stream()
                        .sorted((o1, o2) -> (int) (o1.getTime() - o2.getTime()))
                        .collect(Collectors.toList());
                final int size = spentTimeList.size();
                //数据每个方法的耗时情况
                System.out.println(entry.getKey() + "() -- TP90：" + spentTimeList.get((int) Math.ceil(0.9d * size)).getTime() + "毫秒 -- TP99：" + spentTimeList.get((int) Math.ceil(0.99d * size)).getTime() + "毫秒");
            }
        }, 30, 5, TimeUnit.SECONDS);
    }

    /**
     * 拦截方法，记录访问时间
     * @param invoker
     * @param invocation
     * @return
     * @throws RpcException
     */
    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        long startTime = System.currentTimeMillis();
        Result invoke = null;
        try {
            invoke = invoker.invoke(invocation);
        } catch (RpcException e) {
            e.printStackTrace();
        } finally {
            final long spentTime = System.currentTimeMillis() - startTime;
            DelayQueue<SpentTime> delayQueue = mappedMethodSpentTime.get(invocation.getMethodName());
            if (delayQueue == null) {
                delayQueue = new DelayQueue<>();
            }
            // 记录方法响应时间，过期时间60秒
            delayQueue.put(new SpentTime(spentTime, 60 * 1000));
            mappedMethodSpentTime.put(invocation.getMethodName(), delayQueue);
        }
        return invoke;
    }

    static class SpentTime implements Delayed {
        /* 响应时长 毫秒*/
        private final long time;
        /* 过期时间 */
        private final long expire;

        SpentTime(long time, long delay) {
            this.time = time;
            this.expire = System.currentTimeMillis() + delay;
        }

        public long getTime() {
            return time;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(this.expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }

        @Override
        public int compareTo(Delayed o) {
            final SpentTime o1 = (SpentTime) o;
            return (int) (this.getTime() - o1.getTime());
        }

    }
}
